package com.atguigu.edu.realtime.app.dwd.db;


import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * 支付成功事实表
 */
public class DwdTradePaySucDetail {
    public static void main(String[] args) {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(5);
        //1.3 指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //1.4 设置状态的TTL(传输的延迟 + 业务上的滞后关系)
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(60*15+10));
        //TODO 2.检查点相关的设置(略)
        //TODO 3.从kafka的topic_db主题中读取数据 创建动态表
        tableEnv.executeSql("create table topic_db(" +
                "`database` String,\n" +
                "`table` String,\n" +
                "`type` String,\n" +
                "`data` map<String, String>,\n" +
                "`old` map<String, String>,\n" +
                "`proc_time` as PROCTIME(),\n" +
                "`ts` string,\n" +
                "row_time as TO_TIMESTAMP(FROM_UNIXTIME(cast(ts as bigint))) \n" +
                ")" + MyKafkaUtil.getKafkaDDL("topic_db", "dwd_trade_pay_detail_suc"));
        //tableEnv.executeSql("select * from topic_db").print();
        //TODO 4.过滤出支付成功数据
        Table paymentInfo = tableEnv.sqlQuery("select \n" +
                " `data`['id'] id,\n" +
                " `data`['order_id'] order_id,\n" +
                " `data`['alipay_trade_no'] alipay_trade_no,\n" +
                " `data`['payment_type'] payment_type,\n" +
                " `data`['payment_status'] payment_status,\n" +
                " `data`['callback_content'] callback_content,\n" +
                " `data`['callback_time'] callback_time,\n" +
                "  ts \n" +
                "from topic_db\n" +
                "where `table` = 'payment_info'\n" +
               // "and `type` = 'update'\n" +
                "and data['payment_status']='1602'" );
        tableEnv.createTemporaryView("payment_info", paymentInfo);
        //tableEnv.executeSql("select * from payment_info").print();

        //TODO .从下单事实表中读取下单数据 指定Watermark生成策略以及事件时间字段
        tableEnv.executeSql("create table dwd_trade_order_detail( \n" +
                " id string,\n" +
                " course_id string,\n" +
                " course_name string,\n" +
                " order_id string,\n" +
                " user_id string,\n" +
                " origin_amount string,\n" +
                " coupon_reduce string,\n" +
                " final_amount string,\n" +
                " create_time string,\n" +
                " out_trade_no string,\n" +
                " trade_body string,\n" +
                " session_id string,\n" +
                " province_id string,\n" +
                " source_id string,\n" +
                " ts string,\n" +
                " row_time as TO_TIMESTAMP(FROM_UNIXTIME(cast(ts as bigint))) \n" +
                ")" + MyKafkaUtil.getKafkaDDL("dwd_trade_order_detail", "dwd_trade_pay_detail_suc"));

        //TODO 7.关联上述二张表
        Table resultTable = tableEnv.sqlQuery("select \n" +
                        "    tod.id,\n" +
                        "    tod.course_id,\n" +
                        "    tod.course_name,\n" +
                        "    tod.order_id,\n" +
                        "    tod.user_id,\n" +
                        "    tod.origin_amount,\n" +
                        "    tod.coupon_reduce,\n" +
                        "    tod.final_amount,\n" +
                        "    tod.create_time,\n" +
                        "    tod.out_trade_no,\n" +
                        "    tod.trade_body,\n" +
                        "    tod.session_id,\n" +
                        "    tod.province_id,\n" +
                        "    tod.source_id,\n" +
                        "    pi.alipay_trade_no,\n" +
                        "    pi.payment_type,\n" +
                        "    pi.payment_status,\n" +
                        "    pi.callback_content,\n" +
                        "    pi.callback_time,\n" +
                        "    pi.ts\n" +
                "from dwd_trade_order_detail tod " +
                "join payment_info pi\n" +
                "on tod.order_id = pi.order_id");
        tableEnv.createTemporaryView("result_table", resultTable);

        //TODO 8.将关联的结果写到kafka主题
        //8.1 创建动态表和要写入的主题进行映射
        tableEnv.executeSql("create table dwd_trade_pay_detail_suc(\n" +
                "  id string,\n" +
                "  course_id string,\n" +
                "  course_name string,\n" +
                "  order_id string,\n" +
                "  user_id string,\n" +
                "  origin_amount string,\n" +
                "  coupon_reduce string,\n" +
                "  final_amount string,\n" +
                "  create_time string,\n" +
                "  out_trade_no string,\n" +
                "  trade_body string,\n" +
                "  session_id string,\n" +
                "  province_id string,\n" +
                "  source_id string,\n" +
                "  alipay_trade_no string,\n" +
                "  payment_type string,\n" +
                "  payment_status string,\n" +
                "  callback_content string,\n" +
                "  callback_time string,\n" +
                "  ts string,\n" +
                "  primary key(id) not enforced\n" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_pay_detail_suc"));

        //8.2 写入
        tableEnv.executeSql("insert into dwd_trade_pay_detail_suc select * from result_table");
    }
}
